package cn.ac.iie.ulss.dataredistribution.handler;

import cn.ac.iie.ulss.dataredistribution.commons.GlobalVariables;
import cn.ac.iie.ulss.dataredistribution.commons.RuntimeEnv;
import cn.ac.iie.ulss.dataredistribution.tools.HNode;
import cn.ac.iie.ulss.dataredistribution.tools.MetaStoreClientPool;
import java.io.ByteArrayOutputStream;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicLong;
import org.I0Itec.zkclient.ZkClient;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.hadoop.hive.metastore.api.CreateOperation;
import org.apache.hadoop.hive.metastore.api.CreatePolicy;
import org.apache.hadoop.hive.metastore.api.Node;
import org.apache.hadoop.hive.metastore.api.NodeGroup;
import org.apache.hadoop.hive.metastore.api.SFile;
import org.apache.hadoop.hive.metastore.api.SplitValue;
import org.apache.hadoop.hive.metastore.model.MetaStoreConst;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.util.EntityUtils;
import org.apache.log4j.PropertyConfigurator;
import org.apache.thrift.TException;

/**
 *
 * @author: evan
 * @date: 2014-10-15
 */
public class GetFileFromMetaStore {

    String keyinterval = null;
    HNode node = null;
    String table = null;
    String partT = null;
    String keywords = null;
    String urlEnd = (String) RuntimeEnv.getParam(RuntimeEnv.URL_END);
    SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd-HH:mm:ss");
    SimpleDateFormat dateFormat2 = new SimpleDateFormat("yyyy-MM-dd:HH");
    private static final int ZOOKEEPER_RETRY_INIT = 2000;
    public static AtomicLong HANDLER_CLIENT_SIZE = new AtomicLong(0);
    private static final int CREATEFILE_RETRY_INIT = 2000;
    private static MetaStoreClientPool mscp = (MetaStoreClientPool) RuntimeEnv.getParam(GlobalVariables.METASTORE_CLIENT_POOL);
    int attempSize = 2;
    static org.apache.log4j.Logger logger = null;

    static {
        PropertyConfigurator.configure("log4j.properties");
        logger = org.apache.log4j.Logger.getLogger(GetFileFromMetaStore.class.getName());
    }

    public GetFileFromMetaStore(String keyinterval, HNode node, String table) {
        this.keyinterval = keyinterval;
        this.node = node;
        this.table = table;
        this.partT = node.getPartType();
        this.keywords = node.getKeywords();
    }

    /**
     *
     * check and get a file for the invervla ,if metastore has no such file ,
     * create a new one
     */
    void getFileForInverval() {
        String sendIP = "";
        Long f_id = 0L;
        String road = "";
        Object[] ob = new Object[3];

        String[] keytime = keyinterval.split("\\|");

        Long stime = 0L;
        Long etime = 0L;

        try {
            Date ds = dateFormat.parse(keytime[0]);
            Date de = dateFormat.parse(keytime[1]);
            stime = ds.getTime() / 1000;
            etime = de.getTime() / 1000;
        } catch (ParseException ex) {
            logger.info(ex, ex);
        } catch (Exception ex) {
            logger.error(ex, ex);
        }

        String[] pa = partT.split("\\|");
        long version = Long.parseLong(pa[pa.length - 1]);
        List<SplitValue> list = new ArrayList<SplitValue>();
        for (int j = 1; j <= 2; j++) { //设置一级和二级划分值
            if (j == 1) {
                SplitValue sv1 = new SplitValue();
                sv1.setVerison(version);
                sv1.setLevel(j);//如果是interval分区，设置两个特征值，一个是上限一个是下限
                sv1.setValue("" + stime);
                sv1.setSplitKeyName(keywords.split("\\|")[0]);
                list.add(sv1);

                SplitValue sv2 = new SplitValue();
                sv2.setVerison(version);
                sv2.setLevel(j);//如果是interval分区，设置两个特征值，一个是上限一个是下限
                sv2.setValue("" + etime);
                sv2.setSplitKeyName(keywords.split("\\|")[0]);
                list.add(sv2);
            } else if (j == 2) {
                SplitValue sv = new SplitValue();
                sv.setVerison(version);
                sv.setLevel(j);
                sv.setValue(node.getHashNum() + "-" + node.getName());//设置哈希键值
                sv.setSplitKeyName(keywords.split("\\|")[1]);
                list.add(sv);
            }
        }        
        ob = getUsefulFileFromMetaDB(list);

        sendIP = (String) ob[0];
        f_id = (Long) ob[1];
        road = (String) ob[2];       
        if (sendIP == null || f_id == 0L || road == null || sendIP.equals("") || road.equals("") || sendIP == "" || road == "") {
            logger.info("can't get aomlsf , need to get zk for " + table + " " + keyinterval + " " + node.getName());
            sendIP = "";
            f_id = 0L;
            road = "";
            ob[0] = "";
            ob[1] = 0L;
            ob[2] = "";

            String zkCluster = (String) RuntimeEnv.getParam(RuntimeEnv.ZK_CLUSTER);
            int zkSessionTimeout = 10000;
            int ZKretryInterval = ZOOKEEPER_RETRY_INIT;
            int ZKretryAttempt = 0;

            ZkClient zk = null;
            while (true) {
                try {                    
                    zk = new ZkClient(zkCluster, zkSessionTimeout);                   
                    break;
                } catch (Exception ex) {
                    try {                      
                        Thread.sleep(2000);
                    } catch (Exception ex1) {
                        //
                    }
                }               
            }

            if (!zk.exists("/ulss")) {
                try {
                    zk.createPersistent("/ulss");
                    logger.info("/ulss created");
                } catch (Exception e) {
                    logger.info("root exists : other master has created the /ulss" + e, e);
                }
            }

            if (!zk.exists("/ulss/rd")) {
                try {
                    zk.createPersistent("/ulss/rd");
                    logger.info("/ulss/rd created");
                } catch (Exception e) {
                    logger.info("root exists : other master has created the /ulss/rd" + e, e);
                }
            }

            while (true) {
                try {
                    zk.createEphemeral("/ulss/rd/" + table + node.getName() + "lock", table + node.getName());
                    logger.info("new lock " + table + node.getName() + "lock" + " is created");
                    break;
                } catch (Exception e) {
                    if (ZKretryInterval < 10000) {
                        ZKretryInterval = ZKretryInterval + ZOOKEEPER_RETRY_INIT;
                    } else {
                        ZKretryInterval = 10000;
                    }
                    logger.info("get the zoookeeper lock " + String.format("On retry attempt %d . Sleeping %d seconds.", ++ZKretryAttempt, ZKretryInterval / 1000));
                    try {
                        Thread.sleep(ZKretryInterval);
                    } catch (Exception ex) {
//                        logger.error(ex, ex);
                    }
                    continue;
                }
            }
            ob = getFileFromMetaDB(list);
            sendIP = (String) ob[0];
            f_id = (Long) ob[1];
            road = (String) ob[2];       
            if (sendIP == null || f_id == 0L || road == null || sendIP.equals("") || road.equals("") || sendIP == "" || road == "") {
                sendIP = "";
                f_id = 0L;
                road = "";
                ob[0] = "";
                ob[1] = 0L;
                ob[2] = "";
                while (true) {
                    ob = createNewFile(list);
                    if ("".equals((String) ob[0]) || (Long) ob[1] == 0L || "".equals((String) ob[2])) {
                        try {
                            Thread.sleep(5000);
                        } catch (Exception ex) {
//                                logger.error(ex, ex);
                        }
                        continue;
                    } else {
                        sendIP = (String) ob[0];
                        f_id = (Long) ob[1];
                        road = (String) ob[2];
                        break;
                    }
                }
            }
            zk.close();
            logger.debug("disconnect the zookeeper");
        }

        FileOperator.putFile(table + keyinterval + node.getName() + version, ob);
    }

    /**
     *
     * get file from the metastore before get the zk lock
     */
    private Object[] getUsefulFileFromMetaDB(List<SplitValue> list) {
        Date date = new Date();
        String time = dateFormat2.format(date);
        logger.info(time + " start get a useful file from metadb for " + table + " " + partT.split("\\|")[0] + " " + partT.split("\\|")[1] + " " + keyinterval);      
        Object[] ob = new Object[3];
        String getsendIP = "";
        Long getf_id = 0L;
        String getroad = "";
        List<SFile> lsf = null;
        int attemp = 0;
        while (attemp <= attempSize) {
            MetaStoreClientPool.MetaStoreClient cli = mscp.getClient();
            IMetaStoreClient icli = cli.getHiveClient();
            try {
                lsf = icli.filterTableFiles(partT.split("\\|")[0], partT.split("\\|")[1], list);
                logger.info("get the filelist for " + table + " " + partT.split("\\|")[0] + " " + partT.split("\\|")[1] + " " + keyinterval + " " + lsf);
                cli.release();
                break;
            } catch (TException tex) {
                logger.error("the metastore is bad " + tex, tex);
                try {
                    icli.close();
                } catch (Exception e) {
                }
                mscp.addMetaStoreClient();
            } catch (Exception ex) {
                logger.error("can not get the SFileList for " + table + " " + partT.split("\\|")[0] + " " + partT.split("\\|")[1] + " " + keyinterval + ex, ex);
                cli.release();
                attemp++;
            }
        }

        if (lsf != null && !lsf.isEmpty()) {
            List<SFile> alsf = new ArrayList<SFile>();
            for (SFile s : lsf) {
                if (s.getLoad_status() == MetaStoreConst.MFileLoadStatus.OK && s.getStore_status() == MetaStoreConst.MFileStoreStatus.INCREATE) {
                    alsf.add(s);
                }
            }

            SFile usefulfile = null;
            while (!alsf.isEmpty()) {
                getsendIP = "";
                getf_id = 0L;
                getroad = "";
                usefulfile = new SFile();
                usefulfile.setLength(Long.MAX_VALUE);
                logger.debug("the alsf for " + table + " " + keyinterval + " " + node.getName() + " has " + alsf.size() + " available and open file");
                for (SFile s : alsf) {
                    if (s.getLength() <= usefulfile.getLength()) {
                        usefulfile = s;
                    }
                }

                if (usefulfile.getLocations() == null) {
                    logger.info("the file " + usefulfile + " for " + table + " " + keyinterval + " " + node.getName() + " has no locations");
                    alsf.remove(usefulfile);
                    continue;
                } else {
                    int visit = -1;
                    for (int i = 0; i < usefulfile.getLocations().size(); i++) {
                        if (usefulfile.getLocations().get(i).getVisit_status() == MetaStoreConst.MFileLocationVisitStatus.ONLINE) {
                            visit = i;
                        }
                    }

                    if (visit == -1) {
                        logger.info("the file " + usefulfile + " for " + table + " " + keyinterval + " " + node.getName() + " has no location is online");
                        alsf.remove(usefulfile);
                        continue;
                    } else {
                        try {
                            getsendIP = usefulfile.getLocations().get(visit).getNode_name();//+ rule.getIPList()[0];
                            getf_id = usefulfile.getFid();
                            getroad = usefulfile.getLocations().get(visit).getLocation() + "|" + usefulfile.getLocations().get(visit).getDevid();
                            if (getsendIP == null || getf_id == 0L || getroad == null || getsendIP.equals("") || getroad.equals("") || getsendIP == "" || getroad == "") {
                                alsf.remove(usefulfile);
                                logger.info("there is unvaid information in the file " + usefulfile + " for the topic " + table + " " + keyinterval + " " + node.getName());
                                continue;
                            } else {
                                logger.info("choose the file " + usefulfile + " for the topic " + table + " " + keyinterval + " " + node.getName() + " to the road " + getroad + " to the sendIP " + getsendIP);
                                break;
                            }
                        } catch (Exception e) {
                            alsf.remove(usefulfile);
                            logger.info("there is unvaid information in the file " + usefulfile + " for the topic " + table + " " + keyinterval + " " + node.getName());
                            continue;
                        }
                    }
                }
            }
        } else {
            logger.info(" the SFileList for " + table + " " + keyinterval + " " + node.getName() + " has no available file");
        }
        ob[0] = getsendIP;
        ob[1] = getf_id;
        ob[2] = getroad;
        logger.info(ob.toString());
        return ob;
    }

    private Object[] getFileFromMetaDB(List<SplitValue> list) {
        Date date = new Date();
        String time = dateFormat2.format(date);
        logger.info(time + " start get a useful file from metadb for " + table + " " + partT.split("\\|")[0] + " " + partT.split("\\|")[1] + " " + keyinterval);
        Object[] ob = new Object[3];
        String getsendIP = "";
        Long getf_id = 0L;
        String getroad = "";

        List<SFile> lsf = null;
        int attemp = 0;
        while (attemp <= attempSize) {
            MetaStoreClientPool.MetaStoreClient cli = mscp.getClient();
            IMetaStoreClient icli = cli.getHiveClient();
            try {
                lsf = icli.filterTableFiles(partT.split("\\|")[0], partT.split("\\|")[1], list);
                logger.info("get the filelist for " + table + " " + partT.split("\\|")[0] + " " + partT.split("\\|")[1] + " " + keyinterval + " " + lsf);
                cli.release();
                break;
            } catch (TException tex) {
                logger.error("the metastore is bad " + tex, tex);
                try {
                    icli.close();
                } catch (Exception e) {
                }
                mscp.addMetaStoreClient();
            } catch (Exception ex) {
                logger.error("can not get the SFileList for " + table + " " + partT.split("\\|")[0] + " " + partT.split("\\|")[1] + " " + keyinterval + ex, ex);
                cli.release();
                attemp++;
            }
        }

        if (lsf != null && !lsf.isEmpty()) {
            List<SFile> alsf = new ArrayList<SFile>();
            List<SFile> aclsf = new ArrayList<SFile>();
            for (SFile s : lsf) {
                if (s.getLoad_status() == MetaStoreConst.MFileLoadStatus.OK && s.getStore_status() == MetaStoreConst.MFileStoreStatus.INCREATE) {
                    alsf.add(s);
                } else if (s.getLoad_status() == MetaStoreConst.MFileLoadStatus.OK && s.getStore_status() == MetaStoreConst.MFileStoreStatus.REPLICATED) {
                    aclsf.add(s);
                }
            }

            SFile usefulfile = null;
            while (!alsf.isEmpty()) {
                getsendIP = "";
                getf_id = 0L;
                getroad = "";
                usefulfile = new SFile();
                usefulfile.setLength(Long.MAX_VALUE);
                logger.debug("the alsf for " + table + " " + keyinterval + " " + node.getName() + " has " + alsf.size() + " available and open file");
                for (SFile s : alsf) {
                    if (s.getLength() <= usefulfile.getLength()) {
                        usefulfile = s;
                    }
                }

                if (usefulfile.getLocations() == null) {
                    logger.info("the file " + usefulfile + " for " + table + " " + keyinterval + " " + node.getName() + " has no locations");
                    alsf.remove(usefulfile);
                    continue;
                } else {
                    int visit = -1;
                    for (int i = 0; i < usefulfile.getLocations().size(); i++) {
                        if (usefulfile.getLocations().get(i).getVisit_status() == MetaStoreConst.MFileLocationVisitStatus.ONLINE) {
                            visit = i;
                        }
                    }

                    if (visit == -1) {
                        logger.info("the file " + usefulfile + " for " + table + " " + keyinterval + " " + node.getName() + " has no location is online");
                        alsf.remove(usefulfile);
                        continue;
                    } else {
                        try {
                            getsendIP = usefulfile.getLocations().get(visit).getNode_name();//+ rule.getIPList()[0];
                            getf_id = usefulfile.getFid();
                            getroad = usefulfile.getLocations().get(visit).getLocation() + "|" + usefulfile.getLocations().get(visit).getDevid();
                            if (getsendIP == null || getf_id == 0L || getroad == null || getsendIP.equals("") || getroad.equals("") || getsendIP == "" || getroad == "") {
                                alsf.remove(usefulfile);
                                logger.info("there is unvaid information in the file " + usefulfile + " for the topic " + table + " " + keyinterval + " " + node.getName());
                                continue;
                            } else {
                                logger.info("choose the file " + usefulfile + " for the topic " + table + " " + keyinterval + " " + node.getName() + " to the road " + getroad + " to the sendIP " + getsendIP);
                                break;
                            }
                        } catch (Exception e) {
                            alsf.remove(usefulfile);
                            logger.info("there is unvaid information in the file " + usefulfile + " for the topic " + table + " " + keyinterval + " " + node.getName());
                            continue;
                        }
                    }
                }
            }

            if (alsf.isEmpty()) {
                while (!aclsf.isEmpty()) {
                    getsendIP = "";
                    getf_id = 0L;
                    getroad = "";
                    usefulfile = new SFile();
                    usefulfile.setLength(Long.MAX_VALUE);
                    logger.debug("the alsf for " + table + " " + keyinterval + " " + node.getName() + " has " + alsf.size() + " available and open file");
                    for (SFile s : aclsf) {
                        if (s.getLength() <= usefulfile.getLength()) {
                            usefulfile = s;
                        }
                    }

                    date = new Date();
                    time = dateFormat2.format(date);
                    MetaStoreClientPool.MetaStoreClient cli = mscp.getClient();
                    IMetaStoreClient icli = cli.getHiveClient();
                    try {
                        if (icli.reopen_file(usefulfile.getFid())) {
                            Long oldf_id = usefulfile.getFid();
                            aclsf.remove(usefulfile);
                            usefulfile = icli.get_file_by_id(oldf_id);
                            logger.info(time + " reopen the file for " + table + " " + keyinterval + " " + node.getName() + " " + usefulfile + " successfully");
                        } else {
                            aclsf.remove(usefulfile);
                            usefulfile = null;
                            logger.info(time + " cannotreopen the file for " + table + " " + keyinterval + " " + node.getName() + " " + usefulfile);
                        }
                        cli.release();
                    } catch (TException tex) {
                        aclsf.remove(usefulfile);
                        usefulfile = null;
                        logger.error("the metastore is bad " + tex, tex);
                        try {
                            icli.close();
                        } catch (Exception e) {
                        }
                        mscp.addMetaStoreClient();
                    } catch (Exception ex) {
                        cli.release();
                        aclsf.remove(usefulfile);
                        usefulfile = null;
                        logger.error(time + " cannotreopen the file for " + table + " " + keyinterval + " " + node.getName() + " " + usefulfile + ex, ex);
                    }

                    if (usefulfile == null) {
                        continue;
                    } else {
                        if (usefulfile.getLoad_status() == MetaStoreConst.MFileLoadStatus.OK && usefulfile.getStore_status() == MetaStoreConst.MFileStoreStatus.INCREATE) {
                            if (usefulfile.getLocations() == null) {
                                logger.info("the file " + usefulfile + " for " + table + " " + keyinterval + " " + node.getName() + " has no locations");
                                continue;
                            } else {
                                int visit = -1;
                                for (int i = 0; i < usefulfile.getLocations().size(); i++) {
                                    if (usefulfile.getLocations().get(i).getVisit_status() == MetaStoreConst.MFileLocationVisitStatus.ONLINE) {
                                        visit = i;
                                    }
                                }

                                if (visit == -1) {
                                    logger.info("the file " + usefulfile + " for " + table + " " + keyinterval + " " + node.getName() + " has no location is online");
                                    continue;
                                } else {
                                    try {
                                        getsendIP = usefulfile.getLocations().get(visit).getNode_name();//+ rule.getIPList()[0];
                                        getf_id = usefulfile.getFid();
                                        getroad = usefulfile.getLocations().get(visit).getLocation() + "|" + usefulfile.getLocations().get(visit).getDevid();
                                        if (getsendIP == null || getf_id == 0L || getroad == null || getsendIP.equals("") || getroad.equals("") || getsendIP == "" || getroad == "") {
                                            logger.info("there is unvaid information in the file " + usefulfile + " for the topic " + table + " " + keyinterval + " " + node.getName());
                                            continue;
                                        } else {
                                            logger.info("choose the file " + usefulfile + " for the topic " + table + " " + keyinterval + " " + node.getName() + " to the road " + getroad + " to the sendIP " + getsendIP);
                                            break;
                                        }
                                    } catch (Exception e) {
                                        logger.info("there is unvaid information in the file " + usefulfile + " for the topic " + table + " " + keyinterval + " " + node.getName());
                                        continue;
                                    }
                                }
                            }
                        }
                    }
                }
            }
        } else {
            logger.debug(" the SFileList for " + table + " " + keyinterval + " " + node.getName() + " has no available file");
        }
        ob[0] = getsendIP;
        ob[1] = getf_id;
        ob[2] = getroad;
        return ob;
    }

    /**
     *
     * create a new file by the interval
     *
     */
    private Object[] createNewFile(List<SplitValue> list) {
        String sendIP = "";
        Long f_id = 0L;
        String road = "";
        Object[] ob = new Object[3];
        ob[0] = "";
        ob[1] = 0L;
        ob[2] = "";

        CreatePolicy cp = new CreatePolicy();
        //cp.setOperation(CreateOperation.CREATE_NEW);
        cp.setOperation(CreateOperation.CREATE_NEW_RANDOM);


        SFile sf2 = null;
        List<NodeGroup> lng = null;
        int attemp = 0;
        while (attemp <= attempSize) {
            MetaStoreClientPool.MetaStoreClient cli = mscp.getClient();
            IMetaStoreClient icli = cli.getHiveClient();
            try {
                lng = icli.getTableNodeGroups(partT.split("\\|")[0], partT.split("\\|")[1]);
                cli.release();
                break;
            } catch (TException tex) {
                logger.error("the metastore is bad " + tex, tex);
                try {
                    icli.close();
                } catch (Exception e) {
                }
                mscp.addMetaStoreClient();
            } catch (Exception ex) {
                logger.error("can not get the List<nodeGroup>" + ex, ex);
                cli.release();
                attemp++;
            }
        }

        ArrayList<String> nodeNames = new ArrayList<String>();
        if (lng != null && !lng.isEmpty()) {
            for (NodeGroup ng : lng) {
                for (Node n : ng.getNodes()) {
                    nodeNames.add(n.getNode_name());
                }
            }
        }

        int CFretryInterval = CREATEFILE_RETRY_INIT;
        int CFretryAttempt = 0;

        while (true) {
            MetaStoreClientPool.MetaStoreClient cli = mscp.getClient();
            IMetaStoreClient icli = cli.getHiveClient();
            try {
                Date date = new Date();
                String time = dateFormat2.format(date);
                logger.info(time + " begin to create file for " + table + " " + keyinterval + " " + node.getName() + " " + partT.split("\\|")[0] + " " + partT.split("\\|")[1]);
                sf2 = icli.create_file_by_policy(cp, 2, partT.split("\\|")[0], partT.split("\\|")[1], list);
                cli.release();
            } catch (TException tex) {
                logger.error("the metastore is bad " + tex, tex);
                try {
                    icli.close();
                } catch (Exception e) {
                }
                mscp.addMetaStoreClient();
                continue;
            } catch (Exception ex) {
                logger.error("cannot create the file for " + table + " " + keyinterval + " " + node.getName() + ex, ex);
                cli.release();
                if (CFretryInterval < 10000) {
                    CFretryInterval = CFretryInterval + CREATEFILE_RETRY_INIT;
                } else {
                    CFretryInterval = 10000;
                }
                logger.info("create file for " + table + " " + keyinterval + " " + node.getName() + String.format(" On retry attempt %d . Sleeping %d seconds.", ++CFretryAttempt, CFretryInterval / 1000));
                try {
                    Thread.sleep(CFretryInterval);
                } catch (Exception ex1) {
                }
                continue;
            }
            if (sf2 == null) {
                if (CFretryInterval < 10000) {
                    CFretryInterval = CFretryInterval + CREATEFILE_RETRY_INIT;
                } else {
                    CFretryInterval = 10000;
                }
                logger.info("create file for " + table + " " + keyinterval + " " + node.getName() + String.format(" On retry attempt %d . Sleeping %d seconds.", ++CFretryAttempt, CFretryInterval / 1000));
                try {
                    Thread.sleep(CFretryInterval);
                } catch (Exception ex) {
                }
                continue;
            }
            Date date = new Date();
            String time = dateFormat2.format(date);
            logger.info(time + " create file from metastore successfully for " + table + " " + keyinterval + " " + node.getName());
            String nodeN = sf2.getLocations().get(0).getNode_name();
            if (nodeN == null) {
                if (nodeNames.isEmpty()) {
                    logger.info("there is no node in the nodeGroups for " + partT.split("\\|")[0] + partT.split("\\|")[1]);
                    try {
                        Thread.sleep(5000);
                    } catch (Exception ex) {
                        logger.error(ex, ex);
                    }
                    continue;
                } else {
                    Random r = new Random();
                    int ran = r.nextInt(nodeNames.size());
                    if (nodeNames.get(ran) == null || nodeNames.get(ran).isEmpty()) {
                        logger.info("thres is null node in the nodeGroups for " + partT.split("\\|")[0] + partT.split("\\|")[1]);
                        try {
                            Thread.sleep(5000);
                        } catch (Exception ex) {
                            logger.error(ex, ex);
                        }
                        continue;
                    }
                    sendIP = nodeNames.get(ran);// + rule.getIPList()[0];
                }
            } else {
                sendIP = nodeN;//+ rule.getIPList()[0];
            }
            f_id = sf2.getFid();
            road = sf2.getLocations().get(0).getLocation() + "|" + sf2.getLocations().get(0).getDevid();

            SendCreateFileCMD scfc = new SendCreateFileCMD(sendIP + urlEnd, f_id, road);
            CloseableHttpResponse hp = scfc.send();
            if (hp == null) {
                logger.error("the file  for " + table + " " + keyinterval + " " + node.getName() + " " + sf2 + " can not be set online " + sendIP);
                sendIP = "";
                f_id = 0L;
                road = "";
                setBad(sf2);

                if (CFretryInterval < 10000) {
                    CFretryInterval = CFretryInterval + CREATEFILE_RETRY_INIT;
                } else {
                    CFretryInterval = 10000;
                }
                logger.info("create file for " + table + " " + keyinterval + " " + node.getName() + String.format(" On retry attempt %d . Sleeping %d seconds.", ++CFretryAttempt, CFretryInterval / 1000));
                try {
                    Thread.sleep(CFretryInterval);
                } catch (Exception ex) {
                }
                continue;
            } else {
                logger.debug("sendCreateFileCMD : " + hp.getStatusLine());
                try {
                    try {
                        if (hp.getStatusLine().getStatusCode() != 200) {
                            logger.error("the file  for " + table + " " + keyinterval + " " + node.getName() + " " + sf2 + " can not be set online " + sendIP);
                            sendIP = "";
                            f_id = 0L;
                            road = "";
                            setBad(sf2);

                            if (CFretryInterval < 10000) {
                                CFretryInterval = CFretryInterval + CREATEFILE_RETRY_INIT;
                            } else {
                                CFretryInterval = 10000;
                            }
                            logger.info("create file for " + table + " " + keyinterval + " " + node.getName() + String.format(" On retry attempt %d . Sleeping %d seconds.", ++CFretryAttempt, CFretryInterval / 1000));
                            try {
                                Thread.sleep(CFretryInterval);
                            } catch (Exception ex) {
                            }
                            continue;
                        } else {
                            ByteArrayOutputStream out = new ByteArrayOutputStream();
                            hp.getEntity().writeTo(out);
                            String resonseEn = new String(out.toByteArray());                          
                            if ("0".equals(resonseEn.split("[\n]")[0])) {
                                logger.info(time + " this file " + f_id + " " + table + " " + keyinterval + " " + node.getName() + " " + road + " " + sendIP + " has been set online");
                                ob[0] = sendIP;
                                ob[1] = f_id;
                                ob[2] = road;
                                break;
                            } else {
                                logger.error("the file  for " + table + " " + keyinterval + " " + node.getName() + " " + sf2 + " can not be set online " + hp.getStatusLine() + " " + sendIP + " " + resonseEn);
                                sendIP = "";
                                f_id = 0L;
                                road = "";
                                setBad(sf2);

                                if (CFretryInterval < 10000) {
                                    CFretryInterval = CFretryInterval + CREATEFILE_RETRY_INIT;
                                } else {
                                    CFretryInterval = 10000;
                                }
                                logger.info("create file for " + table + " " + keyinterval + " " + node.getName() + String.format(" On retry attempt %d . Sleeping %d seconds.", ++CFretryAttempt, CFretryInterval / 1000));
                                try {
                                    Thread.sleep(CFretryInterval);
                                } catch (Exception ex) {
//                                        logger.error(ex, ex);
                                }
                                continue;
                            }
                        }
                    } finally {
                        EntityUtils.consume(hp.getEntity());
                        hp.close();
                    }
                } catch (Exception ex) {
                    logger.error("the file  for " + table + " " + keyinterval + " " + node.getName() + " " + sf2 + " can not be set online " + sendIP);
                    sendIP = "";
                    f_id = 0L;
                    road = "";
                    setBad(sf2);

                    if (CFretryInterval < 10000) {
                        CFretryInterval = CFretryInterval + CREATEFILE_RETRY_INIT;
                    } else {
                        CFretryInterval = 10000;
                    }
                    logger.info("create file for " + table + " " + keyinterval + " " + node.getName() + String.format(" On retry attempt %d . Sleeping %d seconds.", ++CFretryAttempt, CFretryInterval / 1000));
                    try {
                        Thread.sleep(CFretryInterval);
                    } catch (Exception ex1) {
                    }
                    continue;
                }
            }
        }
        Date date = new Date();
        String time = dateFormat2.format(date);
        logger.info(time + " new useful file for " + table + " " + keyinterval + " " + node.getName() + " has be created " + sendIP);
        return ob;
    }

    /**
     *
     * set the file bad , it means this file will not be used again
     *
     */
    private void setBad(SFile sf) {
        int attemp = 0;
        while (attemp <= attempSize) {
            MetaStoreClientPool.MetaStoreClient cli = mscp.getClient();
            IMetaStoreClient icli = cli.getHiveClient();
            try {
                icli.set_loadstatus_bad(sf.getFid());
                Date date = new Date();
                String time = dateFormat2.format(date);
                logger.info(time + " has set bad the file for " + table + " " + sf);
                cli.release();
                break;
            } catch (TException tex) {
                logger.error("the metastore is bad " + tex, tex);
                try {
                    icli.close();
                } catch (Exception e) {
                }
                mscp.addMetaStoreClient();
            } catch (Exception ex) {
                logger.error("can not set bad the file " + sf + " " + ex, ex);
                cli.release();
                attemp++;
            }
        }
    }

    public static void setBad(String topic, long f_id) {
        int attemp = 0;
        while (attemp <= 3) {
            MetaStoreClientPool.MetaStoreClient cli = mscp.getClient();
            IMetaStoreClient icli = cli.getHiveClient();
            try {
                icli.set_loadstatus_bad(f_id);
                logger.info("has set bad the file for " + topic + " " + f_id);
                cli.release();
                break;
            } catch (TException tex) {
                logger.error("the metastore is bad " + tex, tex);
                try {
                    icli.close();
                } catch (Exception e) {
                }
                mscp.addMetaStoreClient();
            } catch (Exception ex) {
                logger.error("can not set bad the file " + topic + " " + f_id + " " + ex, ex);
                cli.release();
                attemp++;
            }
        }
    }
}
